/*
  gtask.c
  Description: Cluster Service for GIGI
  Created: 2009.10.26
  Version: 1.0
*/

#ifdef HAVE_CONFIG_H
  #include "config.h"
#endif

#include "gigione.h"
#include "gmpi.h"
#include "gconn.h"
#include <assert.h>

#define THIS_FILE_ERROR_ID  30000

static int _gmpi_parent_node (int id, int * xStep);

static int _gmpi_sync (GIGI_MPI mpi, GIGI_ID service_id,int timeout);

static int _gmpi_bcast_stream (GIGI_MPI mpi, GIGI_STREAM s, int root_id);

static int _gmpi_bcast_buffer (GIGI_MPI mpi,int * buf_size,
                               void ** buf, int root_id);

static GIGI_MPI _gmpi_new (GIGIONE context, GMPI_FUNC_DEF func, GIGI_ID* nodes,
                            int count, int  is_root);


#define FORCE_CLOSE_SOCKET( fd )\
  do { gigi_closesocket (fd, GIGI_TRUE); (fd) = NULL; } while(0)

static GIGI_MPI _gmpi_new (GIGIONE context, GMPI_FUNC_DEF func, GIGI_ID* nodes,
                            int count, int  is_root)
{
  int i, task_size;
  GIGI_MPI mpi;

  task_size = sizeof (struct gigi_mpi_context_t) + sizeof (gmpi_node) * count;
  mpi = (GIGI_MPI) malloc ( task_size );

  if (mpi == NULL)
  {
    LOG_OUT_OF_MEMORY();
    return NULL;
  }

  memset (mpi, 0, task_size );
  mpi->_context        = context;
  mpi->_mpi_func  = func;
  mpi->_size      = count;

  if ( count > 0)
  {
    mpi->_nodes     = (gmpi_node*) ( mpi + 1);
    mpi->_mpi_id    = gigi_gen_id(context);

    for (i = 0; i < count; i ++)
    {
      mpi->_nodes[i]._node_id = nodes[i];
    }
  }

  if (is_root)
  {
    for ( i = 0; i< count; i++)
    {
      if (nodes[i] == context->_my_node_id)
      {
        mpi->_nodes[i]._node_id = nodes[0];
        mpi->_nodes[0]._node_id = nodes[i];
        break;
      }
    }

    assert ( i < count);

    if (i == count)
    {
      free (mpi);
      mpi = NULL;
    }
  }

  return mpi;
};

static int _gmpi_sync (GIGI_MPI mpi, GIGI_ID service_id, int timeout)
{
  int step, x, sync_ack, n, sync_result;
  int my_id;
  int comm_size;
  int parent_id;
  gmpi_node* parent_node;
  gmpi_block task_block_buff, *task_block;
  int task_block_size;

  timeout;

  my_id       = mpi->_id;
  comm_size   = mpi->_size;
  parent_id   = _gmpi_parent_node (my_id, &step);
  parent_node = parent_id < 0 ? NULL : mpi->_nodes + parent_id;

  // Step 1: Check the mpi 
  //
  if ( mpi->_error_code == GIGI_OK)
  {
    mpi->_service = gigi_open_service ( mpi->_context, mpi->_mpi_id, 15);

    if  ( strlen (mpi->_mpi_func->_name) >= sizeof ( ( (gmpi_block *)NULL)->_name) )
    {
      mpi->_error_code = GIGI_E_INTERNAL_ERROR;
      assert ( 0 );
    }
    else if (mpi->_service == NULL )
    {
      mpi->_error_code = GIGI_E_INTERNAL_ERROR;
      assert(0);
    }
  }

  sync_ack = mpi->_error_code;

  // Step 3.1 : send to son nodes
  // Step 3.2 : get ACK from son nodes
  // Step 3.3 : combine the sync_ack with son's SyncACK
  //
  if (mpi->_error_code == GIGI_OK && (my_id + step) < comm_size )
  {
    task_block_size = gmpi_block_size + sizeof (GIGI_ID) * comm_size;

     // prepare task_block
    //
    task_block = comm_size <= GIGI_COUNT_OF (task_block_buff._nodes) 
          ? &task_block_buff 
          : gmpi_block_new (comm_size);
    
    assert ( task_block );

    task_block->_count = HOST_BYTE_INT32 (comm_size);
    task_block->_mpi_id = mpi->_mpi_id;
    BYTE_CHANGE_ID( task_block->_mpi_id );
    
    memset (task_block->_name, 0, sizeof(task_block->_name) );
    strcpy (task_block->_name, mpi->_mpi_func->_name);

    for (x = 0; x < comm_size; x++)
    {
      task_block->_nodes[x] = mpi->_nodes[x]._node_id;
      BYTE_CHANGE_ID( task_block->_nodes[x] );
    };
    
    // send to son node
    //
    for (x = step; x + my_id < comm_size; x <<= 1)
    {
      gmpi_node * pN = mpi->_nodes + x + my_id;

      pN->_fd = gigi_socket (mpi->_context, pN->_node_id, service_id,
                             GIGI_FALSE, NULL);

      if (pN->_fd == NULL )
      {
        // assert ( 0 );
        break;
      }

      task_block->_id = x + my_id;
      
      BYTE_CHANGE_INT32 ( task_block->_id );

      // gmpi_accept::gigi_recv_object( GIGI_TYPE_AUTO_BUF, ...)
      //
      n = gigi_send_object (pN->_fd, GIGI_TYPE_BUFFER, task_block, task_block_size );
    
      if (n < 0)
      {
        FORCE_CLOSE_SOCKET( pN->_fd );
        
        assert (0);

        break;
      }
    }

    if (task_block != &task_block_buff)
    {
      free (task_block);
    };

    // recv ACK from Son
    //

    for (x = step; x + my_id < comm_size; x <<= 1)
    {
      int nAck;
      gmpi_node* pN = mpi->_nodes + x + my_id;

      if (pN->_fd == NULL) 
      { 
        sync_ack = GIGI_E_INTERNAL_ERROR;
        continue;
      }

      // SEND_RECV_001
      //
      n = gigi_recv_object (pN->_fd, GIGI_TYPE_INT32, &nAck, NULL);
      if ( n < 0 )
      {
        nAck = GIGI_E_INTERNAL_ERROR;
      }

      if ( nAck != GIGI_OK )
      {
        sync_ack = nAck;
        FORCE_CLOSE_SOCKET (pN->_fd );
        assert ( 0 );
      }
    }
  }

  // send the confirm message to parent node
  //
  if (parent_node)
  {
    // SEND_RECV_001
    //
    n = gigi_send_object (parent_node->_fd, GIGI_TYPE_INT32, &sync_ack, 0);
    
    if ( n < 0 )
    {
      sync_ack = GIGI_E_INTERNAL_ERROR;
    }

    if (sync_ack != GIGI_OK)
    {
      // CLOSE_SOCKET_001
      //
      FORCE_CLOSE_SOCKET (parent_node->_fd );

      assert (0);

      return sync_ack;
    };
  }

  // Step 4: root node broadcast the nSync Result to all Sons
  //
  // get confirm message from parent node
  //
  if (parent_node)
  {
    // SEND_RECV_002
    //
    n = gigi_recv_object (parent_node->_fd, GIGI_TYPE_INT32, &sync_result, NULL);
    
    if (n < 0 )
    {
      sync_result = GIGI_E_INTERNAL_ERROR;

      FORCE_CLOSE_SOCKET ( parent_node->_fd );

      assert ( 0 );
    }
  }
  else
  {
    sync_result = sync_ack;
  };

  // send the final confirm message for son
  //
  for (x = step; x + my_id < comm_size; x <<= 1)
  {
    gmpi_node* pN = mpi->_nodes + x + my_id;
    
    if (pN->_fd == NULL)
    {
      continue;
    }
    
    // SEND_RECV_002
    //
    n = gigi_send_object (pN->_fd, GIGI_TYPE_INT32, &sync_result, 0);

    if (n < 0 )
    {
      FORCE_CLOSE_SOCKET ( pN->_fd );
        
      sync_result = GIGI_E_INTERNAL_ERROR;
    
      assert ( n >= 0 );
    }
  }

  return sync_result;
};


GIGI_MPI gmpi_create (GIGIONE context, GMPI_FUNC_DEF func, GIGI_ID * nodes, int count)
{
  GIGI_MPI mpi;

  if (context == NULL || count <= 0 || func == NULL)
  {
    return NULL;
  }

  mpi = _gmpi_new (context, func, nodes, count, GIGI_TRUE );

  return mpi;
}

static int _gmpi_run_proxy (GIGI_MPI mpi, void * param)
{
  if (mpi->_mpi_func == NULL  || 
      mpi->_mpi_func->_broker == NULL ||
      mpi->_master == NULL)
  {
    return GIGI_E_INTERNAL_ERROR;
  }

  mpi->_mpi_func->_broker (mpi->_context, mpi->_master, param);

  return GIGI_OK;
}

int gmpi_run (GIGI_MPI mpi, void * param, GIGI_ID service_id, int timeout)
{
  int i, n;

  if (mpi == NULL)
  {
    return GIGI_E_INTERNAL_ERROR;
  };

  if ( mpi->_size == 0 )
  {
    n = _gmpi_run_proxy (mpi, param);
  }
  else
  {
    // prepare the basic field 
    //
    n = _gmpi_sync (mpi, service_id, timeout);

    if (n == GIGI_OK)
    {
      mpi->_mpi_func->_func (mpi, param);
    };
  }

  // close mpi
  //
  for (i =0; i < mpi->_size; i++)
  {
    if (mpi->_nodes[i]._fd)
    {
      gigi_closesocket (mpi->_nodes[i]._fd, GIGI_FALSE);
      mpi->_nodes[i]._fd = NULL;
    }
  };

  if (mpi->_master)
  {
    gigi_closesocket (mpi->_master, GIGI_TRUE);
    mpi->_master = NULL;
  };

  if (mpi->_service)
  {
    gigi_close_service (mpi->_service);
  }

  free (mpi);

  return n;
};

GIGI_SOCKET gmpi_remote_call (const char * func_name, unsigned uIP,
                              int uPORT, GIGI_ID service_id)
{
  GIGI_SOCKET fd;
  gmpi_block task_block;
  int rc;

  if ( strlen (func_name) >= sizeof (task_block._name) )
  {
    return NULL;
  };

  fd = gigi_socket (NULL, GIGI_ID_MAKE(uIP, uPORT, 0), service_id,
                    GIGI_TRUE , NULL);

  if (fd == NULL)
  {
    return fd;
  }

  memset (&task_block, 0, sizeof(task_block) );
  strcpy (task_block._name, func_name);

  rc = gigi_send_object(fd, GIGI_TYPE_BUFFER, &task_block, gmpi_block_size );
  
  if (rc < 0 )
  {
    gigi_closesocket (fd, GIGI_TRUE);
    return NULL;
  }

  return fd;
}

GIGI_MPI gmpi_accept (GIGIONE context, GIGI_ID service_id,int timeout)
{
  gmpi_block * task_block = NULL;  
  GIGI_MPI mpi = NULL;
  GMPI_FUNC_DEF func = NULL;
  int n, nSize, nCheckOK, parent_id = -1;
  GIGI_SOCKET fd;
  GIGI_ID from;
  
  GIGI_HANDLE_CHECK (NULL);

  // Step 1: Recv the mpi block
  //
  fd = gigi_accept (context, &from, service_id, timeout);

  if (fd == NULL)
  {
    return NULL;
  }

  nSize = sizeof (void *);
  n = gigi_recv_object (fd, GIGI_TYPE_PTR_BUF, &task_block, &nSize);
  
  if (n < 0)
  {
    FORCE_CLOSE_SOCKET ( fd );
    assert ( 0 );
    return NULL;
  }

  // Step 2: fatal error check
  //           this type error will not report to parent, 
  //           and close socket directly
  //
  nCheckOK = nSize >= gmpi_block_size;

  if (nCheckOK)
  {
    BYTE_CHANGE_INT32 ( task_block->_count );
    BYTE_CHANGE_INT32 ( task_block->_id    );
    BYTE_CHANGE_ID    ( task_block->_mpi_id);

    if ( task_block->_count == 0)
    {
      nCheckOK = task_block->_id == 0 && task_block->_name[0] != 0; 
    }
    else
    {
      nCheckOK = 
      nSize >= (int) (gmpi_block_size + sizeof (GIGI_ID) * task_block->_count)
      && task_block->_count > 0 
      && task_block->_id    > 0 
      && task_block->_id    < task_block->_count;
    };
  }

  // 
  // this check only for non-proxy mpi
  // 
  if (nCheckOK && task_block->_count > 0)
  {
    int step;
    GIGI_ID aFrom;

    parent_id = _gmpi_parent_node (task_block->_id, &step);
    aFrom = task_block->_nodes[parent_id];
    
    BYTE_CHANGE_ID( aFrom );
    nCheckOK = from == aFrom;
  }

  if (nCheckOK == GIGI_FALSE)
  {
    FORCE_CLOSE_SOCKET (fd );
    free (task_block);
    assert ( 0 );
    return NULL;
  }

  // Step 3: non-Fatal Error Check, this error will report to parent
  //
  task_block->_name[ sizeof (task_block->_name) - 1] = '\0';
  func = gigi_mpi_func_def (context, task_block->_name);

  // Step 4: Check Proxy flag and mpi attribute
  //
  nCheckOK = func != NULL && ( 
        ( task_block->_count == 0 && func->_broker ) 
      ||  ( task_block->_count >= 0 && func->_func)
    );

  if (nCheckOK)
  {
    int i;
    
    for (i = 0; i<task_block->_count && nCheckOK; i++)
    {
      BYTE_CHANGE_ID ( task_block->_nodes[i] );
    }
  };

  // Step 3: create mpi object and initialize the object  
  //
  mpi = nCheckOK 
    ? _gmpi_new (context, func, task_block->_nodes, task_block->_count, GIGI_FALSE) 
    : _gmpi_new (context, NULL, NULL, 0, GIGI_FALSE);
  
  //printf ("recv id %d\n", task_block->_id);

  if (mpi == 0)
  {
    assert( 0 );
    FORCE_CLOSE_SOCKET(fd);
  }
  else if ( mpi->_size > 0 )
  {
    // for Cluster Task
    mpi->_service         = NULL;
    mpi->_mpi_id         = task_block->_mpi_id;
    mpi->_id                  = task_block->_id;
    mpi->_error_code           = nCheckOK ? GIGI_OK : GIGI_E_INTERNAL_ERROR;    
    mpi->_nodes[parent_id]._fd = fd;
  }
  else
  {
    mpi->_master = fd;
    // for Proxy Task mode
  }

  free (task_block);

  return mpi;
};


int gmpi_rank (GIGI_MPI mpi)
{
  if (mpi == NULL)
  {
    return GIGI_E_INVALID_HANDLE;
  }

  return mpi->_id;
}

int gmpi_size (GIGI_MPI mpi)
{
  if (mpi == NULL)
  {
    return GIGI_E_INVALID_HANDLE;
  }

  return mpi->_size;
}

GIGI_ID gmpi_get_node_id (GIGI_MPI mpi, int index)
{
  if (mpi == NULL)
  {
    return 0;
  }

  assert ( index >=0 && index < mpi->_size);
  return mpi->_nodes[index]._node_id;
}

int gmpi_send (GIGI_MPI mpi, int type, void * pData, int nSize, int index)
{
  gmpi_node* pN;

  if (mpi == NULL)
  {
    return GIGI_E_INVALID_HANDLE;
  }

  pN = mpi->_nodes + index;

  assert ( index >=0 && index < mpi->_size);

// create socket for first time
//
  if (pN->_fd == NULL)
  {
    pN->_fd = gigi_socket (mpi->_context, pN->_node_id , mpi->_mpi_id, 0, NULL);
  
    if (pN->_fd == NULL)
    {
      return GIGI_E_INTERNAL_ERROR;
    }
  }

  return gigi_send_object (pN->_fd, type, pData, nSize);
}

static int _gmpi_get_rank_by_node_id (GIGI_MPI mpi, GIGI_ID node_id)
{
  int i;

  for (i =0; i< mpi->_size; i++)
  {
    if (node_id == mpi->_nodes[i]._node_id) return i;
  }

  return GIGI_E_INVALID_ID;
};

static int _gmpi_accept (GIGI_MPI mpi, int index)
{
  assert ( index >= 0 && index < mpi->_size);

  while ( mpi->_nodes[index]._fd == NULL )
  {
    int rank;
    GIGI_ID from;

    GIGI_SOCKET fd = gigi_accept (mpi->_context, &from, mpi->_mpi_id, -1 );
    
    if (fd == NULL) 
    {
      return GIGI_E_INTERNAL_ERROR;
    };

    rank = from == mpi->_nodes[index]._node_id ? index : _gmpi_get_rank_by_node_id (mpi, from);

    if (rank < 0 || mpi->_nodes[rank]._fd != NULL )
    {
      return GIGI_E_INVALID_ID;
    }

    mpi->_nodes[rank]._fd = fd;
  }

  return GIGI_OK;
};

int gmpi_recv (GIGI_MPI mpi, int type, void * pData, int * nSize, int id)
{
  gmpi_node* pN = mpi->_nodes + id;

  assert ( id >=0 && id < mpi->_size);
  
// step 1: prepare the connection to node[id]
//
  if ( pN->_fd == NULL)
  {
    if ( _gmpi_accept (mpi, id ) < 0 )
    {
      return GIGI_E_INTERNAL_ERROR;
    };
  }

  return gigi_recv_object (pN->_fd, type, pData, nSize);
}

static int _gmpi_parent_node (int id, int * step)
{
  int x = 1 << 16;
  
  if (id == 0)
  {
    *step = 1;
    return -1;
  }

  for (; (x & id) == 0; x >>= 1)
  {
  }

  *step = x + x;
  
  return id - x;
};

//  bcast
//  x : 0    1    2     3     4     5     6    7
//  1 : 0->1
//  2 : 0->2 1->3 
//  4 : 0->4 1->5 2->6  3->7 
//  8 : 0->8 1->9 2->10 3->11 4->12 5->13 6->14 7->15
//
//  recv (n)   2^ (m-1) <= n < 2^m
//  parent =  n - 2^ (m-1)
//  
//
int gmpi_bcast (GIGI_MPI mpi, int type, void * buff, int * size, int root_id)
{
  int comm_size, my_id, n, x, parent_id;
  int nSendSize;

  if (mpi == NULL)
  {
    return GIGI_E_INVALID_HANDLE;
  }

  if (type == GIGI_TYPE_STREAM)
  {
    return _gmpi_bcast_stream (mpi, (GIGI_STREAM)buff, root_id);
  }

  comm_size = mpi->_size;
  my_id     = mpi->_id >= root_id ? mpi->_id - root_id : comm_size + mpi->_id - root_id;
  parent_id = _gmpi_parent_node (my_id, &x);
  
  // recv from parent node
  //
  n = parent_id >= 0 ? gmpi_recv (mpi, type, buff, size, (parent_id + root_id) % comm_size ) : GIGI_OK;

  // send to sons
  //
  for ( nSendSize = size ? *size : -1; (x + my_id) < comm_size && n >= 0; x <<= 1 )
  {
    n = gmpi_send (mpi, type, buff, nSendSize, (x + my_id + root_id) % comm_size );
  }

  return n;
}

static void _reduce_sum_int32 (void * a, void * b) 
{ 
  * ( (int *)a)     += * ( (int *)b);     
}

static void _reduce_sum_int64 (void * a, void * b) 
{ 
  * ( (GIGI_UINT64*)a) += * ( (GIGI_UINT64*)b); 
}

static void _reduce_sum_double (void * a, void * b) 
{ 
  * ( (double*)a)  += * ( (double*)b);  
}

static void _reduce_and_int32 (void * a, void * b) 
{ 
  * ( (int *)a)     &= * ( (int *)b);     
}

static void _reduce_and_int64 (void * a, void * b) 
{ 
  * ( (GIGI_UINT64*)a) &= * ( (GIGI_UINT64*)b); 
}

static void _reduce_or_int32 (void * a, void * b) 
{ 
  * ( (int *)a)     |= * ( (int *)b);     
}

static void _reduce_or_int64 (void * a, void * b) 
{ 
  * ( (GIGI_UINT64*)a) |= * ( (GIGI_UINT64*)b); 
}

static void _reduce_xor_int32 (void * a, void * b) 
{ 
  * ( (int *)a)     ^= * ( (int *)b);     
}

static void _reduce_xor_int64 (void * a, void * b) 
{ 
  * ( (GIGI_UINT64*)a) ^= * ( (GIGI_UINT64*)b); 
}

/*
#define GIGI_TYPE_CHAR       1
#define GIGI_TYPE_INT16      2
#define GIGI_TYPE_INT32      3
#define GIGI_TYPE_INT64      4
#define GIGI_TYPE_DOUBLE     5
#define GIGI_TYPE_ID         6

#define GIGI_SUM ( (GIGI_REDUCE_OP)1)
#define GIGI_MAX ( (GIGI_REDUCE_OP)2)
#define GIGI_MIN ( (GIGI_REDUCE_OP)3)
#define GIGI_AND ( (GIGI_REDUCE_OP)4)
#define GIGI_OR  ( (GIGI_REDUCE_OP)5)
#define GIGI_XOR ( (GIGI_REDUCE_OP)6)
*/

GIGI_REDUCE_OP g_InnerReduceFunc[ (__w64 int)GIGI_REDUCE_EOF][GIGI_TYPE_OBJECT_BEGIN] = 
{
  {0}, 
  /*SUM*/{0, 0, _reduce_sum_int32  , _reduce_sum_int64  , _reduce_sum_double, 0}, 
  /*MAX*/{0, 0, 0                  , 0                  , 0, 0 }, 
  /*MIN*/{0, 0, 0                  , 0                  , 0, 0 }, 
  /*AND*/{0, 0, _reduce_and_int32 , _reduce_and_int64 , 0, 0 },
  /*OR */{0, 0, _reduce_or_int32  , _reduce_or_int64  , 0, 0 },
  /*XOR*/{0, 0, _reduce_xor_int32 , _reduce_xor_int64 , 0, 0 },
};

static int g_typeSize[6] = {0, 1,2,4,8,8};

int gmpi_reduce (GIGI_MPI mpi, GIGI_REDUCE_OP op, int type,
                 void * buff, int * count, int root_id)
{
  int n = 0, x, comm_size, my_id, parent_id, step;
  char sPeer[64];
  int nSize;

  // step 1: adjust the op for inner reduce Type
  //
  if ( op < GIGI_REDUCE_EOF )
  {
    if (type < GIGI_TYPE_OBJECT_BEGIN)
    {
      op = g_InnerReduceFunc[ (__w64 int)op ][ type - 1];
      
      if (op == NULL)
      {
        return GIGI_E_INTERNAL_ERROR;
      }
    } 
    else
    {
      return GIGI_E_INTERNAL_ERROR;
    }
  };

  if (type >= GIGI_TYPE_OBJECT_BEGIN)
  {
    return GIGI_E_INTERNAL_ERROR;
  }

  assert( (type & GIGI_TYPE_ARRAY) == 0 );
  assert( (type & 0x4000 ) == 0 );
  assert( type < GIGI_TYPE_OBJECT_BEGIN );
  nSize = g_typeSize [ type ];

  // step 2: reduce 
  //
  comm_size = mpi->_size;

  my_id = mpi->_id >= root_id ? mpi->_id - root_id : comm_size + mpi->_id - root_id;

  parent_id = _gmpi_parent_node (my_id, &step);

  // recv from son nodes
  //
  for ( x = step; x + my_id < comm_size; x <<= 1)
  {
    n = gmpi_recv (mpi, type, sPeer, NULL, (x + my_id + root_id) % comm_size );
    if (n < 0) break;
    op (buff, sPeer);
  }

  // send to parent
  //
  if ( n>=0 && parent_id >=0)
  {
    n = gmpi_send (mpi, type, buff, 0, (parent_id + root_id) % comm_size );
  }

  return n;
};

int gmpi_sync (GIGI_MPI mpi, GIGI_REDUCE_OP op, int type, void * buff, int * count)
{
  int n = gmpi_reduce (mpi, op, type, buff, count, 0);
  
  if ( n >= 0 )
  {
    n = gmpi_bcast (mpi, type, buff, count, 0);
  }

  return n;
};

GIGIONE gmpi_context (GIGI_MPI mpi)
{
  return mpi->_context;
}

static int _gmpi_bcast_buffer (GIGI_MPI mpi, int * buff_size,
                               void ** buff, int root_id)
{
  int n, nOK = GIGI_FALSE, nID = mpi->_id;

  *buff = NULL;

  if (nID == root_id && *buff_size > 0 )
  {
    *buff = malloc ( *buff_size );

    if (*buff == NULL)
    {
      LOG_OUT_OF_MEMORY();
      *buff_size = 0;
    }
  }

  n = gmpi_bcast (mpi, GIGI_TYPE_INT32, buff_size, NULL, root_id);
  CER(n);

  if ( *buff_size == 0)
  {
    return n;
  }

  if ( nID != root_id )
  {
    *buff = malloc ( *buff_size );

    if (*buff == NULL )
    {
      LOG_OUT_OF_MEMORY ();
      *buff_size = 0;
    }
  };

  nOK = *buff != NULL;

  n = gmpi_sync (mpi, GIGI_AND, GIGI_TYPE_INT32, &nOK, NULL);
  CER(n);
  
Cleanup:;
  
  if (!nOK || n < 0)
  {
    if (*buff)
    {
      free (*buff);
      *buff = NULL;
    };

    *buff_size = 0;
  }

  return n;
}

static int _gmpi_bcast_stream (GIGI_MPI mpi, GIGI_STREAM s, int root_id)
{
  int buff_size = 128 * 1024;
  char * buf_p = NULL;
  int nID, nOK, n;

  if ( s == NULL )
  {
    buff_size = 0;
  };

  nID = gmpi_rank (mpi);

  n = _gmpi_bcast_buffer (mpi, &buff_size, (void **)&buf_p, root_id);

  if (n < 0 || buff_size == 0 || s == NULL)
  {
    return n;
  }

  for ( nOK = GIGI_TRUE; nOK; )
  {
    int nReaded = root_id == nID ? gigi_fread (s, buf_p, buff_size) : buff_size;
    
    assert ( nReaded >=0 );

    n = gmpi_bcast (mpi, GIGI_TYPE_BUFFER, buf_p, &nReaded, root_id);
    CER (n);

    if ( nReaded <= 0) 
    {
      nOK = nReaded == 0;
      break;
    }

    if ( root_id != nID)
    {
      nOK = gigi_fwrite (s, buf_p, nReaded) > 0;
    }
  }

  n = gmpi_reduce (mpi, GIGI_AND, GIGI_TYPE_INT32, &nOK, NULL, root_id);
  CER (n);

Cleanup:;
  free (buf_p);
  return n;
};

int gmpi_get_nodes_of_group (GIGI_MPI mpi, GIGI_ID * nodes, int count)
{
  int i;

  if ( count < mpi->_size)
  {
    return -1;
  }

  count = mpi->_size;

  for (i = 0; i < count; i++ )
  {
    nodes [i] = mpi->_nodes[i]._node_id;
  }

  return count;
};

